package com.lagou.rpc.consumer.proxy;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.ZkStrSerializer;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.consumer.client.RpcClient;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.data.Stat;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 客户端代理类-创建代理对象
 * 1.封装request请求对象
 * 2.创建RpcClient对象
 * 3.发送消息
 * 4.返回结果
 */
@Component
@Slf4j
public class RpcClientProxy {


    private ZkClient zkClient;
    private List<String> rpcClientsInfo;
    private Map<String, RpcClient> rpcClientMap;
    private Map<String, RequestTime> rpcClientsPerformance;

    private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);



    @PostConstruct
    private void init(){
        log.info("初始化zkClient连接");
        // 初始化zkClient
        zkClient = new ZkClient("linux121:2181,linux122:2181");
        zkClient.setZkSerializer(new ZkStrSerializer());

        // 读取java-zk节点的数据
        if (!zkClient.exists("/java-zk")) {
            log.error("zk上没有相关配置节点， 无法启动服务器");
            throw new RuntimeException("zk上没有相关配置节点， 无法启动服务器");
        }
        log.info("成功启动zk");

        rpcClientsInfo = zkClient.getChildren("/java-zk")
                .stream()
                .distinct()
                .sorted()
                .collect(Collectors.toList());
        rpcClientMap = new HashMap<>();
        rpcClientsPerformance = new HashMap<>();
        for (String info : rpcClientsInfo) {
            rpcClientMap.put(info, getRpcClient(info));
        }

        zkClient.subscribeChildChanges("/java-zk", new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                List<String> currentServers = currentChilds
                        .stream()
                        .distinct()
                        .sorted()
                        .collect(Collectors.toList());
                // 获取失效的服务， 断开连接
                Set<String> invalidServers = new HashSet<>(rpcClientsInfo);
                invalidServers.removeAll(currentServers);
                for (String invalidServer : invalidServers) {
                    RpcClient rpcClient = rpcClientMap.get(invalidServer);
                    rpcClient.close();
                    rpcClientMap.remove(invalidServer);
                    System.out.println("服务" + invalidServer + "已经失效， 关闭连接");
                }


                // 获取新增的服务， 建立连接
                Set<String> newServers = new HashSet<>(currentServers);
                newServers.removeAll(rpcClientsInfo);
                for (String newServer : newServers) {
                    RpcClient rpcClient = getRpcClient(newServer);
                    System.out.println("服务" + newServer + "连接成功");
                    rpcClientMap.put(newServer, rpcClient);
                }

                rpcClientsInfo = currentServers;
            }
        });

        executorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                System.out.println(rpcClientsPerformance);

                // 将响应时间数据上传
                for (Map.Entry<String, RequestTime> entry : rpcClientsPerformance.entrySet()) {
                    long current = System.currentTimeMillis();
                    if (current - entry.getValue().getTime() <= 5000L){
                        // 当记录时间是5秒内才上传
                        zkClient.writeData("/java-zk/" + entry.getKey(), entry.getValue().getCost());
                    }
                }

                // 读取数据
                for (String info : rpcClientsInfo) {
                    long current = System.currentTimeMillis();
                    Stat stat = new Stat();
                    String tmp = zkClient.readData("/java-zk/" + info, stat);
                    if (tmp == null || tmp.trim().isEmpty() || current - stat.getMtime() > 5000L){
                        // 当服务端记录的更新时间是5秒外， 删除节点的内容, 同时移除本地的内容
                        zkClient.writeData("/java-zk/" + info, "");
                        rpcClientsPerformance.remove(info);
                    }else{
                        // 当服务端记录的更新时间是5秒内， 更新本地列表
                        Long cost = Long.parseLong(tmp);
                        RequestTime requestTime = new RequestTime();
                        requestTime.setTime(stat.getMtime());
                        requestTime.setCost(cost);
                        rpcClientsPerformance.put(info, requestTime);;
                    }

                }
                System.out.println(rpcClientsPerformance);


            }
        }, 0, 5, TimeUnit.SECONDS);

    }

    @PreDestroy
    public void destroy(){
        zkClient.close();
    }


    private RpcClient getRpcClient(String info){
        String[] split = info.split(":");
        return new RpcClient(split[0], Integer.parseInt(split[1]));
    }

    public Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        //1.封装request请求对象
                        RpcRequest rpcRequest = new RpcRequest();
                        rpcRequest.setRequestId(UUID.randomUUID().toString());
                        rpcRequest.setClassName(method.getDeclaringClass().getName());
                        rpcRequest.setMethodName(method.getName());
                        rpcRequest.setParameterTypes(method.getParameterTypes());
                        rpcRequest.setParameters(args);
                        //2.创建RpcClient对象
                        RpcClient rpcClient = getRpcClient();
                        try {
                            //3.发送消息
                            long before = System.currentTimeMillis();
                            Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));
                            long after = System.currentTimeMillis();
                            RequestTime requestTime = rpcClientsPerformance.get(rpcClient.getAddress());
                            if (requestTime == null){
                                requestTime = new RequestTime();
                                rpcClientsPerformance.put(rpcClient.getAddress(), requestTime);
                            }
                            requestTime.setCost(after - before);
                            requestTime.setTime(before);


                            RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                            if (rpcResponse.getError() != null) {
                                throw new RuntimeException(rpcResponse.getError());
                            }
                            //4.返回结果
                            Object result = rpcResponse.getResult();
                            return JSON.parseObject(result.toString(), method.getReturnType());
                        } catch (Exception e) {
                            throw e;
                        }

                    }
                });
    }


    private synchronized RpcClient getRpcClient(){
        // 所有的服务器都有请求后， 再根据响应时间查找最优服务器
        if (rpcClientsPerformance.size() == rpcClientsInfo.size()){
            // 如果存在响应记录， 获取响应最快的服务器
            Map.Entry<String, RequestTime> min
                    = rpcClientsPerformance.entrySet().stream().min((o1, o2) -> (int) (o1.getValue().getCost() - o2.getValue().getCost()))
                    .get();
            System.out.println("获取最有服务器:" + min.getKey() + "其他服务器信息： " + rpcClientsPerformance);
            return rpcClientMap.get(min.getKey());
        }

        // 不存在则进行随机取服务器
        String key = rpcClientsInfo.get((int) (Math.random() * rpcClientsInfo.size()));
        return rpcClientMap.get(key);
    }
}
